package com.raylu.app

import com.alibaba.fastjson.{JSON, JSONObject}
import com.alibaba.fastjson.serializer.SerializeConfig
import com.raylu.bean.{OrderDetail, OrderInfo, OrderWide}
import com.raylu.util.{MyEsUtil, MyKafkaUtil, OffsetManagerUtil, PropertiesUtil, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import redis.clients.jedis.Jedis

import java.time.{LocalDate, Period}
import java.util
import java.util.Properties
import scala.collection.mutable.ListBuffer

/**
 *
 * Description:
 *
 * Create by lucienoz on 2021/12/5.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
object DwOrderApp {
	def main(args : Array[String]) : Unit = {
		/**
		 * 加工实时流数据装配订单信息数据
		 */

		val sparkConf : SparkConf = new SparkConf ().setAppName ( "dw_dau_app" ).setMaster ( "local[*]" )
		val ssc = new StreamingContext ( sparkConf, Seconds ( 5 ) )
		val orderInfoTopic : String = "DWD_ORDER_INFO_I"
		val orderDetailTopic : String = "DWD_ORDER_DETAIL_I"
		val groupId : String = "dw_order_group"


		val orderInfoOffset : Map[TopicPartition, Long] = OffsetManagerUtil.getOffset ( orderInfoTopic, groupId )
		val orderDetailOffset : Map[TopicPartition, Long] = OffsetManagerUtil.getOffset ( orderDetailTopic, groupId )

		//防止redis记录用户首次日活，而数据在进入es时发生错误，导致该用户首次日活无法被redis拦截


		var orderInfoKafkaDStream : InputDStream[ConsumerRecord[String, String]] = null
		if (orderInfoOffset.isEmpty && orderInfoOffset == null) orderInfoKafkaDStream = MyKafkaUtil.getKafkaStreamOfConsumer ( orderInfoTopic, ssc, groupId )
		else orderInfoKafkaDStream = MyKafkaUtil.getKafkaStreamOfConsumer ( orderInfoTopic, ssc, groupId, orderInfoOffset )

		var orderDetailKafkaDStream : InputDStream[ConsumerRecord[String, String]] = null
		if (orderDetailOffset.isEmpty && orderDetailOffset == null) orderDetailKafkaDStream = MyKafkaUtil.getKafkaStreamOfConsumer ( orderDetailTopic, ssc, groupId )
		else orderDetailKafkaDStream = MyKafkaUtil.getKafkaStreamOfConsumer ( orderDetailTopic, ssc, groupId, orderDetailOffset )


		var orderInfoOffsetRanges : Array[OffsetRange] = null
		val orderInfoKafkaDStream2 : DStream[ConsumerRecord[String, String]] = orderInfoKafkaDStream.transform { RDD =>
			orderInfoOffsetRanges = RDD.asInstanceOf[HasOffsetRanges].offsetRanges
			RDD
		}

		var orderDetailOffsetRanges : Array[OffsetRange] = null
		val orderDetailKafkaDStream2 : DStream[ConsumerRecord[String, String]] = orderDetailKafkaDStream.transform { RDD =>
			orderDetailOffsetRanges = RDD.asInstanceOf[HasOffsetRanges].offsetRanges
			RDD
		}

		val orderInfoDS : DStream[(String, OrderInfo)] = orderInfoKafkaDStream2.map { record =>
			val orderInfoJSONStr : String = record.value ()
			val orderInfo : OrderInfo = JSON.parseObject ( orderInfoJSONStr, classOf[OrderInfo] )
			val pk : String = orderInfo.id
			(pk, orderInfo)
		}

		val orderDetailDS : DStream[(String, OrderDetail)] = orderDetailKafkaDStream2.map { record =>
			val orderDetailJSONStr : String = record.value ()
			val orderDetail : OrderDetail = JSON.parseObject ( orderDetailJSONStr, classOf[OrderDetail] )
			val fk : String = orderDetail.order_id
			(fk, orderDetail)
		}

		//orderInfoDS 与 orderDetailDS 进行关联

		val fullJoinDS : DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoDS.fullOuterJoin ( orderDetailDS )

		fullJoinDS.foreachRDD { RDD =>
			RDD.foreachPartition {
				fullJoinOptionIter =>
					val jedis : Jedis = RedisUtil.getJedisFromPool ()
					val jedisDIM : Jedis = RedisUtil.getJedisFromPool ()
					val properties : Properties = PropertiesUtil.load ( "config.properties" )
					jedis.select ( properties.getProperty ( "redis.order.db" ).toInt )
					jedisDIM.select ( properties.getProperty ( "redis.dimtable.db" ).toInt )

					var dt : String = null

					val listBuffer : ListBuffer[OrderWide] = ListBuffer[OrderWide]()
					for ((key, (orderInfoOp, orderDetailOp)) <- fullJoinOptionIter) {
						if (dt == null) {
							val dateTime : String = if (!orderInfoOp.isEmpty) orderInfoOp.get.create_time else orderDetailOp.get.create_time
							dt = dateTime.substring ( 0, 10 )
						}

						//将主表 orderInfoOp 数据缓存到redis
						if (!orderInfoOp.isEmpty) {
							val orderInfo : OrderInfo = orderInfoOp.get
							jedis.set ( s"ORDERINFO:${key}", JSON.toJSONString ( orderInfo, new SerializeConfig ( true ) ) )
						}

						//判断orderInfoOp 与 orderDetailOp 是否关联成功，若成功则将
						//数据发送给ES
						if (!orderInfoOp.isEmpty && !orderDetailOp.isEmpty) {
							val orderInfo : OrderInfo = orderInfoOp.get
							val orderDetail : OrderDetail = orderDetailOp.get
							val orderWide : OrderWide = new OrderWide ( orderInfo, orderDetail )
							//Merge
							println ( "user_id====>" + orderWide.user_id )
							val userInfoJSONStr : String = jedisDIM.get ( s"DIM:user_info:${orderWide.user_id}" )
							//							println("userInfoJSONStr======>"+userInfoJSONStr)
							if (userInfoJSONStr == null) println ( "user_id====>" + orderWide.user_id + " is not exists on redis" )
							else {

								val userInfoJSONObj : JSONObject = JSON.parseObject ( userInfoJSONStr )
								//							println ("userInfoJSONObj=======>"+ userInfoJSONObj.getClass )
								val birthday : String = userInfoJSONObj.getString ( "birthday" )
								if (birthday == null) {
									val birthdayLocalDate : LocalDate = LocalDate.parse ( birthday )
									val now : LocalDate = LocalDate.now ()
									val period : Period = Period.between ( birthdayLocalDate, now )
									val age : Int = period.getYears
									orderWide.user_age = age
								}
								orderWide.user_gender = userInfoJSONObj.getString ( "gender" )
							}

							val baseProvinceJSONStr : String = jedisDIM.get ( s"DIM:base_province:${orderWide.province_id}" )
							val baseProvinceJSONObject : JSONObject = JSON.parseObject ( baseProvinceJSONStr )
							orderWide.province_3166_2 = baseProvinceJSONObject.getString ( "iso_3166_2" )
							orderWide.province_area_code = baseProvinceJSONObject.getString ( "area_code" )
							orderWide.province_iso_code = baseProvinceJSONObject.getString ( "iso_code" )
							orderWide.province_name = baseProvinceJSONObject.getString ( "name" )

							listBuffer.append ( orderWide )
						}

						//当orderInfoOp 与 orderDetailOp未关联成功时则将orderDetailOp数据
						//与redis中OrderInfo数据匹配，匹配到将数据放入ES中，匹配不到的orderDetailOp放入redis中
						if (orderInfoOp.isEmpty && !orderDetailOp.isEmpty) {
							val orderDetail : OrderDetail = orderDetailOp.get
							val orderInfoJSONObjStr : String = jedis.get ( s"ORDERINFO:${key}" )
							if (orderInfoJSONObjStr == null) {
								//将orderDetail数据放入redis
								jedis.sadd ( s"ORDERDETAIL:${key}", JSON.toJSONString ( orderDetail, new SerializeConfig ( true ) ) )
							} else {
								//匹配成功将数据存入ES
								val orderInfo : OrderInfo = JSON.parseObject ( orderInfoJSONObjStr, classOf[OrderInfo] )
								val orderWide = new OrderWide ( orderInfo, orderDetail )
								//Merge from DIM
								val userInfoJSONStr : String = jedisDIM.get ( s"DIM:user_info:${orderWide.user_id}" )
								if (userInfoJSONStr == null) println ( "user_id====>" + orderWide.user_id + " is not exists on redis" )
								else {

									val userInfoJSONObj : JSONObject = JSON.parseObject ( userInfoJSONStr )
									//							println ("userInfoJSONObj=======>"+ userInfoJSONObj.getClass )
									val birthday : String = userInfoJSONObj.getString ( "birthday" )
									if (birthday == null) {
										val birthdayLocalDate : LocalDate = LocalDate.parse ( birthday )
										val now : LocalDate = LocalDate.now ()
										val period : Period = Period.between ( birthdayLocalDate, now )
										val age : Int = period.getYears
										orderWide.user_age = age
									}
									orderWide.user_gender = userInfoJSONObj.getString ( "gender" )
								}

								val baseProvinceJSONStr : String = jedisDIM.get ( s"DIM:base_province:${orderWide.province_id}" )
								val baseProvinceJSONObject : JSONObject = JSON.parseObject ( baseProvinceJSONStr )
								orderWide.province_3166_2 = baseProvinceJSONObject.getString ( "iso_3166_2" )
								orderWide.province_area_code = baseProvinceJSONObject.getString ( "area_code" )
								orderWide.province_iso_code = baseProvinceJSONObject.getString ( "iso_code" )
								orderWide.province_name = baseProvinceJSONObject.getString ( "name" )

								listBuffer.append ( orderWide )
							}


						}

						//将orderInfoOp数据与
						//在redis中orderDetail数据进行关联，关联到则将关联数据存入ES中
						if (!orderInfoOp.isEmpty && orderDetailOp.isEmpty) {
							val orderInfo : OrderInfo = orderInfoOp.get
							val orderDetailStrUtilSet : util.Set[String] = jedis.smembers ( s"ORDERDETAIL:${key}" )
							import scala.collection.JavaConverters._
							orderDetailStrUtilSet.asScala.map { orderDetailStr =>
								val orderDetail : OrderDetail = JSON.parseObject ( orderDetailStr, classOf[OrderDetail] )
								orderDetail
							}.foreach { orderDetail =>
								val orderWide = new OrderWide ( orderInfo, orderDetail )
								//Merge from DIM
								val userInfoJSONStr : String = jedisDIM.get ( s"DIM:user_info:${orderWide.user_id}" )
								if (userInfoJSONStr == null) println ( "user_id====>" + orderWide.user_id + " is not exists on redis" )
								else {

									val userInfoJSONObj : JSONObject = JSON.parseObject ( userInfoJSONStr )
									//							println ("userInfoJSONObj=======>"+ userInfoJSONObj.getClass )
									val birthday : String = userInfoJSONObj.getString ( "birthday" )
									if (birthday == null) {
										val birthdayLocalDate : LocalDate = LocalDate.parse ( birthday )
										val now : LocalDate = LocalDate.now ()
										val period : Period = Period.between ( birthdayLocalDate, now )
										val age : Int = period.getYears
										orderWide.user_age = age
									}
									orderWide.user_gender = userInfoJSONObj.getString ( "gender" )
								}

								val baseProvinceJSONStr : String = jedisDIM.get ( s"DIM:base_province:${orderWide.province_id}" )
								val baseProvinceJSONObject : JSONObject = JSON.parseObject ( baseProvinceJSONStr )
								orderWide.province_3166_2 = baseProvinceJSONObject.getString ( "iso_3166_2" )
								orderWide.province_area_code = baseProvinceJSONObject.getString ( "area_code" )
								orderWide.province_iso_code = baseProvinceJSONObject.getString ( "iso_code" )
								orderWide.province_name = baseProvinceJSONObject.getString ( "name" )

								listBuffer.append ( orderWide )

							}

						}


					}

					//ES save operations
					var flag : String = null
					if (listBuffer.size > 0) {
						flag = MyEsUtil.bulkSave ( s"dws_order_wide_${dt}", listBuffer.map { orderWide =>
							(orderWide.id, orderWide)
						}.toList )
					}

					println ( "ES==>" + flag )

					jedis.close ()
					jedisDIM.close ()
			}

			OffsetManagerUtil.saveOffset ( orderInfoTopic, groupId, orderInfoOffsetRanges )
			OffsetManagerUtil.saveOffset ( orderDetailTopic, groupId, orderDetailOffsetRanges )
		}

		ssc.start ()
		ssc.awaitTermination ()


	}

}
